package org.streets.database.internal;

import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Locale;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.streets.commons.util.JDBCUtils;
import org.streets.commons.util.StringUtils;
import org.streets.database.ExecMode;
import org.streets.database.Limits;
import org.streets.database.PV;
import org.streets.database.RecordHandler;
import org.streets.database.RecordHandlerManager;
import org.streets.database.SQLBatcher;
import org.streets.database.SQLConnection;
import org.streets.database.SQLNestedException;
import org.streets.database.SQLTransaction;
import org.streets.database.SQLTransactionException;
import org.streets.database.SQLUtils;
import org.streets.database.dialect.Dialect;

/**
 *
 * JDBC connection  for commonsensible database operation.
 *
 * @version 1.1.0
 * 		NEW named parameter implementation
 * 		PV parameter helper support 	
 * @version 1.0.0 
 * 		remove NpSql, NpUtils
 * 		Optimized name parameter parse, 
 * 		Add handler to map resultset to java bean
 * 		Add SQLBatcher for batch operation
 * @version 0.8.5 add NpSql suport named parameter sql
 * @version 0.8.0 modified obtainXXX -> prepare method
 * @version 0.7.0
 * @version 0.6.0 uuid indentified statament
 * @version 0.5.0 hashtable maintained statements
 * @author dzb
 */
public class SQLConnectionImpl implements SQLConnection {

    private static final Log LOG = LogFactory.getLog(SQLConnectionImpl.class.getName());
    
    private Connection _connection;    
    private Dialect _dialect;
    private RecordHandlerManager _handlerManager;
    
    public SQLConnectionImpl() {
    	
    }
        
    public SQLConnectionImpl(Connection connection) {
    	this._connection = connection;
    }
    
    public SQLConnectionImpl(Connection connection, Dialect dialect) {
    	this._connection = connection;
    	this._dialect = dialect;
    }

    public SQLConnectionImpl(Connection connection, Dialect dialect, RecordHandlerManager manager) {
    	this._connection = connection;
    	this._dialect = dialect;
    	this._handlerManager = manager;
    }
    
	/**
	 * Prepares the connection to be used by query.<br>
	 * Gets the connection from data source if it's the case.
	 */
	private void check() throws SQLNestedException {
		
		if (_connection == null) {
			throw new SQLNestedException("Connection is NULL, No avalible connection.");
		}
			
		try {
			if (_connection.isClosed()) {
				throw new SQLNestedException("Connection is closed already.");
			}				
		} catch (SQLException ex) {
			throw new SQLNestedException("Fetch Connection error.", ex);
		}
	}
	
	private ResultSet conv(ResultSet rs) throws SQLException {
		Locale locale = Locale.getDefault ();
		try{
			//NOTE: 此处缺省使用Oracle的RowSet实现 如使用其他数据库 需更改
			Locale.setDefault (Locale.US);
			oracle.jdbc.rowset.OracleCachedRowSet crs = new oracle.jdbc.rowset.OracleCachedRowSet();
	        crs.populate(rs);
	        return crs;
		} finally	{
			Locale.setDefault (locale);
		}
	}
	
	/**
	 * Return a {@link List} with elements of the given type by iterating
	 * through the given {@link ResultSet} which is expected to be valid and
	 * transforming iterated {@link ResultSet} with the given
	 * {@link RecordHandler}
	 * 
	 * @param <T>
	 *            the type of elements in the resulted {@link List}
	 * @param rs
	 *            the {@link ResultSet} to be processed
	 * @param handler
	 *            the {@link RecordHandler} which will be used
	 * @return a {@link List} obtained by transforming the given
	 *         {@link ResultSet} using the given {@link RecordHandler}
	 * @throws Exception
	 */
	private <T> List<T> transform(ResultSet rs, RecordHandler<T> handler) throws Exception {

		final List<T> result = new ArrayList<T>();
		// transform the result set using the given result set handler
		while (rs.next()) {
			result.add(handler.mapping(rs, this));
		}

		return result;
	}

	public Connection connection() {
		return _connection;
	}

	public Dialect dialect() {
		return _dialect;
	}

    public void setDialect(Dialect dialect) {
    	this._dialect = dialect;
    }

    public SQLTransaction beginTransaction() {    	
    	check();    	
    	if (isConnectionInTransaction()) {
    		throw new SQLTransactionException("Current Connection already in transaction");
    	}    	
        try {
            _connection.setAutoCommit(false);
            _connection.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
        } catch (SQLException e) {
            LOG.warn("Start transaction failed.", e);
            JDBCUtils.close(_connection);
            throw new SQLTransactionException("Start transaction failed.", e);
        }        
    	return new SQLTransactionImpl(_connection);
    }

    public boolean isConnectionInTransaction() {
    	boolean in_trans;
        try {
        	in_trans  = ! _connection.getAutoCommit();
        } catch (SQLException e) {
        	LOG.warn("Could not fetch connection autocommit status");
        	in_trans = false;
        }
        return in_trans;
    }

    public void close() {
    	// 查看连接是否在未决事务中
        if (isConnectionInTransaction()) {
        	throw new SQLNestedException("Current connection is in trasaction. Can not close.");
        }
        JDBCUtils.close(_connection);
    }

    public final SQLBatcher prepare(String paramSql) {
    	check();
    	String sql = SQLUtils.parsePreparedSQL(paramSql);
    	String[] names = SQLUtils.parseParameterNames(paramSql);
    	PreparedStatement st = null;
    	try {
            st = _connection.prepareStatement(
            		sql,
                    ResultSet.TYPE_FORWARD_ONLY,
                    ResultSet.CONCUR_READ_ONLY);
            
            return new SQLBatcherImpl(st, sql, names);
        } catch (SQLException ex) {
        	JDBCUtils.logError(ex, LOG);
        	JDBCUtils.close(st);
        	throw new SQLNestedException("Error occurs on SQL:"+paramSql, ex);
        }    	
    }

    public SQLBatcher prepare(String paramSql, String paramTypeExpr) {
    	check();
    	String sql =SQLUtils.parsePreparedSQL(paramSql);
    	String[] names = SQLUtils.parseParameterNames(paramSql);
    	int[] types = SQLUtils.parseJDBCTypes(names, paramTypeExpr);
        try {
        	PreparedStatement st = _connection.prepareStatement(
        			sql,
                    ResultSet.TYPE_FORWARD_ONLY,
                    ResultSet.CONCUR_READ_ONLY);
        	
        	return new SQLBatcherImpl(st, sql, names, types);
        } catch (SQLException ex) {
            LOG.error("DBERROR OCCURS: \\r\\n CODE-" + ex.getErrorCode()
                    + "  MESSAGE: " + ex.getMessage()+"\\r\\n"
                    + "  REF SQL: "+sql);
        	throw new SQLNestedException(ex);
        }
    }

    public ResultSet query(String paramSql, Limits limits, Object[] params) {
    	check();
    	PreparedStatement pstmt = null;
    	try {
    		paramSql = SQLUtils.parsePreparedSQL(paramSql);
    		paramSql = SQLUtils.buildDialectLimitQuery(paramSql, limits, _dialect);
	    	pstmt = _connection.prepareStatement(paramSql);
	    	Object[] values = SQLUtils.buildDialectParamBindValues(params, limits, _dialect);
	    	JDBCUtils.bindParameters(pstmt, values);
	    	ResultSet rs = pstmt.executeQuery();
	    	return conv(rs);
    	} catch (SQLException ex) {
            JDBCUtils.logError(ex, LOG);
            throw new SQLNestedException("Error occurs on SQL:"+paramSql, ex);
    	} finally {
    		JDBCUtils.close(pstmt);
    	}
    }

    public ResultSet query(String paramSql, Limits limits, PV... params) {
    	check();
    	String sql = SQLUtils.parsePreparedSQL(paramSql);
    	sql = SQLUtils.buildDialectLimitQuery(sql, limits, _dialect);
    	String[] names = SQLUtils.parseParameterNames(paramSql);
    	PreparedStatement pstmt = null;
    	try {
	    	pstmt = _connection.prepareStatement(sql);
	    	Object[] values = SQLUtils.buildDialectParamBindValues(names, params, limits, _dialect);
	    	JDBCUtils.bindParameters(pstmt, values);
	    	ResultSet rs = pstmt.executeQuery();
	    	return conv(rs);	    
    	} catch (SQLException ex) {
            JDBCUtils.logError(ex, LOG);
            throw new SQLNestedException("Error occurs on SQL:"+paramSql, ex);
    	} finally {
    		JDBCUtils.close(pstmt);
    	}
    }
    	
	public <T> List<T> query(String paramSql, Class<T> clazz, Limits limits, Object[] params) {
		
		RecordHandler<T> handler = _handlerManager.getHandler(clazz);
		if (handler == null) {
			throw new SQLNestedException("Can not find "+clazz+"'s ResultSetHandler from HandlerManager.");
		}
		/* TODO ...
		// Nested Check
		if (handler instanceof AbstractRecordHandler) {
			AbstractRecordHandler arh = (AbstractRecordHandler) handler;
			if (arh.getNestLimit() > 0 && arh.getNestCount() > arh.getNestLimit()) {
				return null;
			}
		}		
		*/
		ResultSet rs = this.query(paramSql, limits, params);
		try {
			return transform(rs, handler);
		} catch (Exception ex) {
			throw new SQLNestedException(paramSql, ex);
		} finally {
			JDBCUtils.close(rs);
		}
	}

	public <T> List<T> query(String paramSql, Class<T> clazz, Limits limits, PV... params) {
		
		RecordHandler<T> handler = _handlerManager.getHandler(clazz);		
		if (handler == null) {
			throw new SQLNestedException("Can not find "+clazz+"'s ResultSetHandler from HandlerManager.");
		}
		/* TODO ...
		// Nest count limit
		if (handler instanceof AbstractRecordHandler) {
			AbstractRecordHandler arh = (AbstractRecordHandler) handler;
			if (arh.getNestLimit() > 0 && arh.getNestCount() > arh.getNestLimit()) {
				return null;
			}
		}		
		*/
		ResultSet rs = this.query(paramSql, limits, params);
		try {
			return transform(rs, handler);
		} catch (Exception ex) {
			throw new SQLNestedException(paramSql, ex);
		} finally {
			JDBCUtils.close(rs);
		}
	}
	/**
	 * 
	 */    
	public <T> List<T> query(String paramSql, RecordHandler<T> handler, Limits limits, Object[] params) {		
		ResultSet rs = this.query(paramSql, limits, params);
		try {
			return transform(rs, handler);
		} catch (Exception ex) {
			throw new SQLNestedException(paramSql, ex);
		} finally {
			JDBCUtils.close(rs);
		}
	}
	/**
	 * 
	 */
    public <T> List<T> query(String paramSql, RecordHandler<T> handler, Limits limits, PV... params) {
    	
    	ResultSet rs = this.query(paramSql, limits, params);
    	try {
    		return transform(rs, handler);
    	} catch (Exception ex) {
    		throw new SQLNestedException(paramSql, ex);
    	} finally {
    		JDBCUtils.close(rs);
    	}
    } 
    /**
     * 取第一列数据，按实际类型返回
     */
    @SuppressWarnings("unchecked")
	public <T> List<T> column(String paramSql, Limits limits, Object[] params) {
    	paramSql = SQLUtils.parsePreparedSQL(paramSql);
		ResultSet rs = this.query(paramSql, limits, params);
		// the result
		final List<T> result = new ArrayList<T>();
		// process the result set
		try {
			while (rs.next()) {
				result.add((T) rs.getObject(1));
			}
		} catch (Exception ex) {
			throw new SQLNestedException(paramSql, ex);
		} finally {
			JDBCUtils.close(rs);
		}
		return result;
	}
	/**
	 * 取第一列数据，按实际类型返回
	 */
    @SuppressWarnings("unchecked")
	public <T> List<T> column(String paramSql, Limits limits, PV... params) {
		
		ResultSet rs = this.query(paramSql, limits, params);
		// the result
		final List<T> result = new ArrayList<T>();
		// process the result set
		try {
			while (rs.next()) {
				result.add((T) rs.getObject(1));
			}
		} catch (Exception ex) {
			throw new SQLNestedException(paramSql, ex);
		}
		return result;
	}
    

	@SuppressWarnings("unchecked")
	public <T> T column(String paramSql, Object[] params) {

		ResultSet resultSet = this.query(paramSql, Limits.none(), params);		
		try {
			if (resultSet.next()) {
				T ret = (T) resultSet.getObject(1);
				if (resultSet.next()) {
					throw new SQLNestedException("SQL:["+paramSql+"] result is not unique ");
				}
				return ret;
			}
			return null;
		} catch (Exception e) {
			throw new SQLNestedException(e);
		} finally {
			JDBCUtils.close(resultSet);
		}
	}

	@SuppressWarnings("unchecked")
	public <T> T column(String paramSql, PV... params) {

		ResultSet resultSet = this.query(paramSql, Limits.none(), params);
		try {
			if (resultSet.next()) {
				T ret = (T) resultSet.getObject(1);
				if (resultSet.next()) {
					throw new SQLNestedException("SQL:["+paramSql+"] result is not unique ");
				}
				return ret;
			}
			return null;
		} catch (Exception e) {
			throw new SQLNestedException(e);
		} finally {
			JDBCUtils.close(resultSet);
		}
	}
	
    /*
     * 批量执行单条或批量SQL语句，如果是多条，语句间以";"分割
	 * @see org.streets.database.Executor#execute(java.lang.String)
	 */
    public synchronized int[] execute(String sqls) {
    	check();
        int[] result = { -1 };
        Statement stmt = null;
        //String cur_sql= "";
        try {
        	stmt = _connection.createStatement();
	        if (sqls.indexOf(";") > 0) {
		        String[] sql_array = sqls.split("[;]");
	            for (int i = 0; i < sql_array.length; i++) {
	                if (StringUtils.isNotBlank(sql_array[i])) {
	                    stmt.addBatch(sql_array[i]);
	                }
	            }	            
	            result = stmt.executeBatch();
	        } else {
	        	boolean b = stmt.execute(sqls);
	        	if (b) {
	        		result = new int[] {stmt.getUpdateCount()};
	        	} else {
	        		result = new int[] {-1};
	        	}	        	
	        }
            return result;
        } catch (SQLException ex) {
        	JDBCUtils.logError(ex, LOG);
        	throw new SQLNestedException("execute sqls:["+sqls+"] error", ex);
        } finally {
        	JDBCUtils.close(stmt);
        }
    }

    public synchronized int execute(String paramSql, Object[] values) {
    	check();
    	PreparedStatement pstmt = null;
    	try {
    		paramSql = SQLUtils.parsePreparedSQL(paramSql);
	    	pstmt = _connection.prepareStatement(paramSql);
	    	JDBCUtils.bindParameters(pstmt, values);
	    	return pstmt.executeUpdate();
    	} catch (SQLException ex) {
            JDBCUtils.logError(ex, LOG);
            throw new SQLNestedException("execute sql: ["+paramSql+"] error!", ex);
    	} finally {
    		JDBCUtils.close(pstmt);
    	}
    }

    
    public synchronized int execute(String paramSql, PV... params ) {
    	check();
    	String sql = SQLUtils.parsePreparedSQL(paramSql);
    	String[] names = SQLUtils.parseParameterNames(paramSql);
    	PreparedStatement pstmt = null;
    	try {
	    	pstmt = _connection.prepareStatement(sql);
	    	Object[] values = SQLUtils.buildNamedParamValues(names, params);
	    	JDBCUtils.bindParameters(pstmt, values);
	    	return pstmt.executeUpdate();	    	
    	} catch (SQLException ex) {
            JDBCUtils.logError(ex, LOG);
            throw new SQLNestedException("execute sql: ["+paramSql+"] error!", ex);
    	} finally {
    		JDBCUtils.close(pstmt);
    	}
    }
    /**
     * 
     */
    public synchronized <T> int execute(T bean, ExecMode mode) {
		
		String sql;
		
		switch (mode) {
		case CREATE:
			sql = SQLUtils.buildJPAInsertSQL(bean.getClass());
			break;	
		case UPDATE: 
			sql = SQLUtils.buildJPAUpdateSQL(bean.getClass());
			break;
		case DELETE: 
			sql = SQLUtils.buildJPADeleteSQL(bean.getClass());
			break;		
		default:
			throw new SQLNestedException("Not support mode ["+mode+"] error!");
		}
		
		String[] names = SQLUtils.parseParameterNames(sql); 
		Object[] params;
		
		try {
			params = SQLUtils.buildJPAFieldValues(names, bean);
		} catch (Exception e) {
			throw new SQLNestedException("Build execute params in mode ["+mode+"]  sql for ["+bean.getClass()+"] error ! ", e);
		}
		
		return this.execute(sql, params);	
	}

    public synchronized <T> int[] execute(List<T> beans, ExecMode mode) {
    		
		if (beans.size() == 0) return new int[]{0};
		
		String sql;
		
		switch (mode) {		
		case CREATE:
			sql = SQLUtils.buildJPAInsertSQL(beans.get(0).getClass());
			break;
		case UPDATE:
			sql = SQLUtils.buildJPAUpdateSQL(beans.get(0).getClass());
			break;
		case DELETE:
			sql = SQLUtils.buildJPADeleteSQL(beans.get(0).getClass());
			break;
		default:
			throw new SQLNestedException("Not support mode ["+mode+"] error!");		
		}
		
		List<Object[]> params_list = new ArrayList<Object[]>(beans.size());
		String[] names = SQLUtils.parseParameterNames(sql);
		try {
			for (T bean : beans) {
				Object[] params = SQLUtils.buildJPAFieldValues(names, bean);
				params_list.add(params);
			}
		} catch (Exception ex) {
			throw new SQLNestedException("Build execute statement in mode ["+mode+"] sql for ["+beans.get(0).getClass()+"] error ! ", ex);
		}
		SQLBatcher batcher = this.prepare(sql);
		// user batcher to execute list params
		for (Object[] params : params_list) {
			batcher.add(params);	
		}		
		return batcher.execute();		
	}
    
    public <T> int[] delete(Class<T> clazz, Object... id) {
    	
    	if (id == null) return new int[]{0};
    	
    	String sql = SQLUtils.buildJPADeleteSQL(clazz);
    	
    	SQLBatcher batcher = this.prepare(sql);
    	for (Object o : id) {
    		batcher.add(new Object[] {o});
    	}
    	return batcher.execute();
    }
   
//
//	public <K, V> Map<K, V> executeQuery(String paramSql, RecordHandler<Map<K, V>> handler, Limits limits, PV... params) {
//
//		// get the result set
//		ResultSet resultSet = this.query(paramSql, limits, params);
//		
//		// process the result set
//		try {
//			return handler.mapping(resultSet);
//		} catch (Exception e) {
//			throw new SQLNestedException(
//					"Error processing the result set for query=" + paramSql, e);
//		} 
//	}
//	
//	public <K, V> Map<K, V> executeQuery(String paramSql, RecordHandler<Map<K, V>> handler, Limits limits, Object[] params) {
//
//		// get the result set
//		ResultSet resultSet = this.query(paramSql, limits, params);
//		
//		// process the result set
//		try {
//			return handler.mapping(resultSet);
//		} catch (Exception e) {
//			throw new SQLNestedException(
//					"Error processing the result set for query=" + paramSql, e);
//		} 
//	}
//    
//	public <T> Collection<T> executeQuery(String paramSql, RecordHandler<Collection<T>> handler, Limits limits, Object[] params) {
//
//		ResultSet resultSet = this.query(paramSql, limits, params);
//		// process the result set
//		try {
//			return handler.mapping(resultSet);
//		} catch (Exception e) {
//			throw new SQLNestedException(
//					"Error processing the result set for query=" + paramSql, e);
//		}
//	}
//
//	public <T> Collection<T> executeQuery(String namedParamSql, RecordHandler<Collection<T>> handler, Limits limits, PV... params) {
//
//		ResultSet resultSet = this.query(namedParamSql, limits, params);
//		// process the result set
//		try {
//			return handler.mapping(resultSet);
//		} catch (Exception e) {
//			throw new SQLNestedException(
//					"Error processing the result set for query=" + namedParamSql, e);
//		}
//	}

	public <T> T unique(String paramSql, RecordHandler<T> handler, Object[] params) {

//		if (handler instanceof AbstractRecordHandler) {
//			((AbstractRecordHandler)handler).inc();
//		}
		
		ResultSet rs = this.query(paramSql, Limits.none(), params);
		
		T result = null;
		
		try {			
			if (rs.next()) {
				result = (T) handler.mapping(rs, this);
				if (rs.next()) {
					throw new SQLNestedException("SQL:["+paramSql+"] result is not unique ");
				}
			}
			return result;			
		} catch (Exception e) {			
			throw new SQLNestedException("Unique sql: ["+paramSql+"] error!", e);			
		} finally {			
			JDBCUtils.close(rs);
		}
	}


	public <T> T unique(Class<T> clazz, Object id) {
		
		RecordHandler<T> handler = _handlerManager.getHandler(clazz);
		if (handler == null) {
			throw new SQLNestedException("Can not find "+clazz+"'s ResultSetHandler from HandlerManager.");
		}
		String sql = SQLUtils.buildJPARetriveSQL(clazz);		
		return unique(sql, handler, PV.of(id));
	}
	
	public <T> T unique(String paramSql, Class<T> clazz, Object[] params) {
		
		List<T> results = query(paramSql, clazz, Limits.none(), params);
		if (results.size() > 0)
			return results.get(0);
		else 
			return null;
	}

    @SuppressWarnings("unchecked")
    public boolean callProc(String proc, Object[] params) {

		check();
		
		CallableStatement stmt = null;		
		try {
			// prepare call
			proc = SQLUtils.parsePreparedSQL(proc);
			stmt = _connection.prepareCall(proc);
			// set arguments			
			if (params.length == 1 && params[0] instanceof Collection) {
				int i = 1;
				for (Object parameter : (Collection<?>) params[0]) {
					stmt.setObject(i, parameter);
					i++;
				}
			} else {
				for (int i = 0; i < params.length; i++) {
					stmt.setObject(i + 1, params[i]);					 
				}
			}
			// execute
			return stmt.execute();

		} catch (Exception e) {
			
			throw new SQLNestedException("Error executing call : " + proc + ", with parameters : " + params, e);
			
		} finally {
			
			JDBCUtils.close(stmt);
		}
    }

	public long count(String sql, String column) {
		Object v = column("select count(1) ( distinct " + column + ") from ("
				+ sql + ") t");
		if (v instanceof java.math.BigDecimal) {
			return ((java.math.BigDecimal) v).longValue();
		} else {
			return (Long) v;
		}
	}
    
    public long count(String sql, PV... params) {     
        Object v = column("select count(1) from (" + sql + ") t", params);
        if (v instanceof java.math.BigDecimal) {
            return ((java.math.BigDecimal) v).longValue();
        } else {
            return (Long) v;
        }
    }
	
	public long count(String paramSql, Object[] params) {
		Object v = column("select count(1) from (" + paramSql + ") t", params);
		if (v instanceof java.math.BigDecimal) {
			return ((java.math.BigDecimal) v).longValue();
		} else {
			return (Long) v;
		}
	}
}
